package com.czk.utils;

import com.czk.constant.Constants;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;

import java.io.IOException;

/**
 * 1 创建命名空间
 * 2 判断表是否存在
 * 3 创建表
 *
 * @Author:ChenZhangKun
 * @Date: 2021/12/2 16:22
 */
public class HbaseUtil {
    private HbaseUtil() {
    }

    /**
     * 创建命名空间
     *
     * @param namespace
     * @throws IOException
     */
    public static void createNameSpace(String namespace) throws IOException {
        // 获取链接
        Connection connection = ConnectionFactory.createConnection(Constants.configuration);
        // 获取admin对象
        Admin admin = connection.getAdmin();
        // 创建命名空间描述器
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
        // 创建命名空间
        admin.createNamespace(namespaceDescriptor);
        // 关闭
        admin.close();
        connection.close();
    }

    /**
     * 表是否存在
     *
     * @param tableName
     * @return
     * @throws IOException
     */
    private static boolean isTableExist(String tableName) throws IOException {
        // 获取链接
        Connection connection = ConnectionFactory.createConnection(Constants.configuration);
        // 获取admin对象
        Admin admin = connection.getAdmin();
        boolean tableEnabled = admin.tableExists(TableName.valueOf(tableName));
        // 关闭
        admin.close();
        connection.close();
        return tableEnabled;
    }

    public static void createTable(String tableName, int versions, String... cfs) throws IOException {
        // 1 判断是否传入了列族信息
        if (cfs.length == 0) {
            System.out.println("请设置列族信息");
            return;
        }
        // 2 判断表是否存在
        boolean tableExist = isTableExist(tableName);
        if (tableExist) {
            System.out.println("表存在");
            return;
        }
        // 3 获取connection
        Connection connection = ConnectionFactory.createConnection(Constants.configuration);
        // 4 获取admin对象
        Admin admin = connection.getAdmin();
        // 5 创建表描述器
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
        // 6 循环添加列族信息
        for (String cf : cfs) {
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);
            // 7 设置版本
            hColumnDescriptor.setMaxVersions(versions);
            hTableDescriptor.addFamily(hColumnDescriptor);
        }
        // 8 创建表
        admin.createTable(hTableDescriptor);
        // 关闭
        admin.close();
        connection.close();
    }

    public static void close(Connection connection, Table... tables) {
        try {
            if (connection != null) {
                connection.close();
            }
            if (tables != null) {
                for (Table table : tables) {
                    table.close();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
